#include "consumer/client/RouteInfoDownloader.h"
#include "consumer/client/ConsumerMediator.h"
#include "codec/RequestBuilder.h"

namespace zeroplus::mq::consumer
{
    RouteInfoDownloader::RouteInfoDownloader(std::shared_ptr<ConsumerMediator> mediator, std::string uri)
        : mediator_(mediator)
    {
        client_ = std::make_unique<ClientNode>(uri);
    }

    RouteInfoDownloader::~RouteInfoDownloader()
    {
    }

    void RouteInfoDownloader::Start()
    {
        loop_.set_name("RouteInfoDownloader");
        loop_.async_run();
        loop_.wait_for_quit();
    }

    void RouteInfoDownloader::SetTopicName(const std::string &topic)
    {
        auto handler = [this, topic]()
        {
            transport::QueryRouteInfoResponse queryResponse = SendQueryRouteInfoRequest(topic);
            if (queryResponse.has_topic_route_info() &&
                queryResponse.topic_route_info().broker_data_list_size() > 0)
            {
                routeinfo::TopicRouteInfo topicRouteInfo = queryResponse.topic_route_info();
                auto [queueData, brokerData, res] = selector_.SelectRouteInfo(topicRouteInfo, RANDOM);
                auto brokerAddrIt = brokerData.broker_addrs().find(BrokerState::MASTER);
                if (brokerAddrIt != brokerData.broker_addrs().end())
                {
                    mediator_->Notify(Connect, Bytes::from_string(brokerAddrIt->second));
                    mediator_->Notify(Subscribe, Bytes::from_string(topic));
                    loop_.quit();
                    return;
                }
            }
            std::this_thread::sleep_for(std::chrono::milliseconds(1000)); // 等待 1 秒后重试
            loop_.post_task([this, topic]()
                            { SetTopicName(topic); }); // 重新提交任务
        };
        loop_.post_task(handler);
    }

    transport::QueryRouteInfoResponse RouteInfoDownloader::SendQueryRouteInfoRequest(const std::string &topic)
    {
        transport::QueryRouteInfoRequest request;
        request.set_topic(topic);

        RequestBuilder<transport::RequestCode::QUERY_ROUTEINFO_BY_TOPIC,
                       transport::QueryRouteInfoRequest,
                       transport::QueryRouteInfoResponse>
            builder(client_);
        auto [responseCode, response] = builder.InvokeSync(request);
        return response;
    }
}
